package org.ykhl.zheer.dispatcher;

import com.google.common.base.Preconditions;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.clients.consumer.internals.NoOpConsumerRebalanceListener;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.common.errors.WakeupException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import org.tempuri.IHisApplay;
import org.tempuri.MediInfoHis;
import org.tempuri2.HisService;
import org.ykhl.interfaces.commons.AdapterInterface;
import org.ykhl.model.RetInfo;
import org.ykhl.mq.commons.*;
import org.ykhl.mq.dispatcher.MqMessageDispatcher;

import javax.xml.namespace.QName;
import java.io.PrintWriter;
import java.io.StringWriter;
import java.net.URL;
import java.util.*;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

@Service
public class MqMessageDispatcherKafkaImpl extends MqMessageDispatcher<BaseMessageKafkaImpl> {
	private static final Logger LOG = LoggerFactory.getLogger(MqMessageDispatcherKafkaImpl.class);
	private static final int CONSUME_THREAD_CNT = 10;

	private String url;
	private String method;//方法
	private String domainName;//域名

	private String url2;
	private String method2;//方法



	private String domainName2;//域名



	private IHisApplay apply =null;
	private HisService hisService=null;

	private ExecutorService fetcherService;
	private ExecutorService consumeService;
	private List<StoppableInfiniteRunnable> fetchRunners;

	private Producer<String, String> producer;

	//存放院内对接实现的类
	private List<String> impClassList = new ArrayList<>();

	//存放编号与实现类之间的关系
	private Map<String,AdapterInterface> fidToimpClass = new HashMap<>();
	//存放fid与交易类型的关系
	private Map<String,String> fidToTradeType = new HashMap<>();
	private String hospitalId;

	private AdapterInterface adapterInterface;
	private RetInfo retInfo = new RetInfo();
	public  void setHospitalId(String hospitalId) {
		this.hospitalId = hospitalId;
	}

	public void setUrl(String url){
		this.url = url;
	}
	public void setMethod(String method){
		this.method = method;
	}
	public void setDomainName(String domainName){
		this.domainName = domainName;
	}

	public void setUrl2(String url2) {
		this.url2 = url2;
	}

	public void setMethod2(String method2) {
		this.method2 = method2;
	}

	public void setDomainName2(String domainName2) {
		this.domainName2 = domainName2;
	}
	@Override
	protected void doInit() throws Exception {
		LOG.info("mq-hospital start to init...");

		Preconditions.checkNotNull(mqNameServerAddress, "mqNameServerAddress not set");
		Preconditions.checkNotNull(hospitalName, "hospitalName not set");
		Preconditions.checkNotNull(useSsl, "useSsl not set");

		LOG.info("mqNameServerAddress: " + mqNameServerAddress);
		LOG.info("mqConsumerGroupName: " + mqConsumerGroupName);
		LOG.info("mqProducerGroupName: " + mqProducerGroupName);
		LOG.info("useSsl: " + useSsl);
		LOG.info("hospitalName:" + hospitalName);
		LOG.info("hospitalId:" + hospitalId);

		fetcherService = Executors.newCachedThreadPool(new NamedThreadFactory("fetcherService"));
		consumeService = Utils.newBlockingFixedThreadPool(
				CONSUME_THREAD_CNT, 200 * CONSUME_THREAD_CNT,
				new NamedThreadFactory("consumeService"));

		//用于发送消息
		producer = new KafkaProducer<>(Utils.genKafkaProducerProps(mqNameServerAddress, useSsl));
		fetchRunners = new ArrayList<>();
		for (int i = 0; i < 2; i++) {	// just one fetcher thread for now
			// TODO: multi-thread way: instantiate consumers one for each thread.
		//接收app发送过来的请求
			AppServerFetchRunner appServerFetchRunner = new AppServerFetchRunner();
			LOG.info(appServerFetchRunner.toString());
			fetchRunners.add(appServerFetchRunner);
			fetcherService.submit(appServerFetchRunner);
		}
		LOG.info("init - done");
		logSender.serviceStarted();
	}

	private boolean shutdownServiceAndWait(ExecutorService service, String serviceName) {
		service.shutdown();
		try {
			if (!service.awaitTermination(30, TimeUnit.SECONDS)) {
				LOG.error("wait " + serviceName + " to terminate, timeout");
				service.shutdownNow();
			}
		} catch (InterruptedException e) {
			LOG.error("", e);
			logSender.error(e);
			service.shutdownNow();
		}

		return service.isShutdown();
	}
	
	@Override
	protected void doDestroy() {

		for (StoppableInfiniteRunnable runner : fetchRunners)
			runner.stop();

		shutdownServiceAndWait(fetcherService, "fetcherService");
		mqMessageTransformer.destroy();
		shutdownServiceAndWait(consumeService, "consumeService");
		producer.close();

		LOG.info("destroy - done");
		logSender.serviceShutdown();
	}

	@Override
	public boolean sendToAppserver(BaseMessageKafkaImpl message) {
		producer.send(message.toProducerRecord());

		LOG.debug("sendToAppserver - message sent, topic: " + message.getTopic() +
				"\n, app server id: " + message.getAppServerId());
		return true;
	}


	//用于处理收到app发送过来的消息
	private class AppServerFetchRunner extends StoppableInfiniteRunnable {
		private Consumer<String, String> consumer;

		@Override
		public void stop(){
			super.stop();
			consumer.wakeup();
		}

		@Override
		public void beforeRun() {
			//TODO: 考虑设置每次接受数据的大小，增加吞吐量
			consumer = new KafkaConsumer<>(
					Utils.genKafkaConsumerProps(mqNameServerAddress, mqConsumerGroupName, useSsl));
			//subscribe topics that match "mq-from-appserver-topic-*"
			consumer.subscribe(Utils.genKafkaSubscribePatternByPrefix(
					Constants.MQ_FROM_APPSERVER_TOPIC_PREFIX),
					new NoOpConsumerRebalanceListener());
		}

		@Override
		public void doRun() {
			try {
				ConsumerRecords<String, String> records = consumer.poll(2000);
//				int i=0;
				for (ConsumerRecord<String, String> record : records) {
					LOG.debug("get " + records.count() + " messages at one time");
					//表示收到restful 过来的请求
					logSender.messageReceivedFromRestful(new BaseMessageKafkaImpl(record));
					//往院内系统发送消息，返回后进行规则文件的转换
					consumeService.submit(new HospitalMsgConsumeRunner(record));
//					i++;
//					if(i==20){
//						consumer.commitAsync();
//						i=0;
//					}
				}
				consumer.commitAsync();
			} catch (WakeupException e) {
				// wake up by other threads, going to stop, do nothing, just recheck state will do
			} catch (CommitFailedException e) {
				LOG.error("failed to commit, commit next time then", e);
				consumer.commitSync();
			} catch (Exception e){
				LOG.error("unknown exception during fetching from hospital", e);
			}
		}
		@Override
		public void afterRun() {
			LOG.info("starting to close the consumer for fetching from mq hospital");
			consumer.close();
		}
	}

	private class HospitalMsgConsumeRunner implements Callable<Void>{

		private final BaseMessageKafkaImpl baseMessage;

		private HospitalMsgConsumeRunner(ConsumerRecord<String, String> record) {
			baseMessage = new BaseMessageKafkaImpl(record);
		}

		@Override
		public Void call() throws Exception {
			try {
				//把消息发往院内系统
				String fid = null;
				String tradeType = baseMessage.getFid();
				if (tradeType != null || !tradeType.equals("")){
					tradeType = fidToTradeType.get(tradeType);
					if(tradeType ==null ||tradeType.equals("")){
						LOG.error("fid is not support,fid value:" + tradeType+",check methodInfo.properties!");
						return null;
					}
				}else{
					return null;
				}
				if (baseMessage.getMessageType() != BaseMessage.MessageType.REQUEST ||
						baseMessage.getRequestId() == null) {
					LOG.warn("HospitalMsgConsumeRunner - not a request message: " + baseMessage);
					LOG.warn("HospitalMsgConsumeRunner - req type: " + baseMessage.getMessageType());
					LOG.warn("HospitalMsgConsumeRunner - req id : " + baseMessage.getRequestId());
					return null;
				}
				fid = baseMessage.getFid();
				if (fid == null || baseMessage.getMessageBody() == null){
					LOG.error("HospitalMsgConsumeRunner - fid or messageBody npe!");
					return null;
				}
				baseMessage.setHospitalId(hospitalId);
				baseMessage.setHospitalName(hospitalName);
				//发送到院内系统
				LOG.debug("HospitalMsgConsumeRunner - fid: " + fid + " request start to Hospital"  + "--Parame :" + baseMessage.getMessageBody());
				HashMap<String,Object> param = new HashMap<String, Object>();
				param.put("fid",fid);
				param.put("requestParam" ,baseMessage.getMessageBody());
				param.put("apply" ,apply);
				param.put("tradeType",tradeType);
				param.put("hisService",hisService);

				//记录发送到Hospital
				logSender.messageSentToHospital(baseMessage);
				String resp =getHosMessageInfo(param);
				if(resp !=null){
					//院内返回的内容消息体
					baseMessage.setMessageBody(resp);
					LOG.debug("Hospital return baseMessage result size:" +baseMessage.toString().length());
					logSender.messageReceivedFromHospital(baseMessage);//日志信息记录消息院内已经返回

					//进行规则转换
					mqMessageTransformer.transformHospitalMsgAsync(baseMessage, MqMessageDispatcherKafkaImpl.this);

				}
			} catch (InterruptedException e) {
				LOG.error("", e);
			} catch (Exception e) {
				connWS();
				LOG.error("consumeMsg failed, fatal !!!", e);
				//fatal("consumeMsg failed, fatal !!!", e);
			}
			return null;
		}
	}
	/**
	 * connect hospital His
	 */
	private void connWS() {

		LOG.debug("Init His webservice");

		try {
			MediInfoHis ss = new MediInfoHis((new URL(url)), (new QName(domainName, method)));
			apply = ss.getWebPoint();

		}catch (Exception e){
			LOG.error("Init webservice error " + e.getMessage());
		}
	}
	private void connHS(){
		LOG.debug("Init His2 webservice");
		try{
			URL wsdlURL = HisService.WSDL_LOCATION;
			hisService = new HisService(wsdlURL,new QName(domainName2,method2));
		}catch (Exception e){
			LOG.error("Init His2 webservice error " + e.getMessage());
		}
	}
	protected  void hospitalInit(){
		try{
			connWS();//连接His
			connHS();
		}catch (Exception e){
			StringWriter errors = new StringWriter();
			e.printStackTrace(new PrintWriter(errors));
			LOG.error("hospitalInit connWS is error,message:" + errors.toString());
		}
		LOG.debug("HospitalAdaptor hospitalInit start!");
		ResourceBundle resource = ResourceBundle.getBundle("ykhlAdaptor/consumer");//consumer为属性文件名，放在ykhlAdaptor下，如果是放在src下，直接用consumer即可
		String implClassValue = resource.getString("implClass");
		LOG.debug("implClassValue:" + implClassValue);
		if(implClassValue != "" || implClassValue != null){
			String[] impClasss =  implClassValue.split(";");
			for (int i=0;i<impClasss.length;i++){
				String impClass = impClasss[i];
				impClassList.add(impClass);
				try{
					AdapterInterface adapterInterface = (AdapterInterface) Class.forName(impClass).newInstance();
					if(i==0){
						fidToimpClass.put("default",adapterInterface);//设置默认实现类为第一个实现类
					}
					String classTofidValue = resource.getString(impClass);
					if(!classTofidValue.equals("") || classTofidValue != null){
						String[]  fids =  classTofidValue.split(";");
						for(String fid:fids){
							fidToimpClass.put(fid,adapterInterface);
						}
					}
				}catch (Exception e){
					LOG.warn(e.getMessage());
				}
			}
		}

		ResourceBundle method = ResourceBundle.getBundle("zheer/methodInfo");
		Enumeration<String>	keys =method.getKeys();
		while (keys.hasMoreElements()) {
			// 获取每一个名称
			String key = (String) keys.nextElement();
			// 利用已得到的名称通过资源包获得相应的值
			String value = method.getString(key);
			fidToTradeType.put(key,value);
		}

		LOG.debug("HospitalAdaptor hospitalInit end!");
	}
	/**
	 *
	 * @param params  fid  requestParam
	 * @return
	 */
	public  String getHosMessageInfo(Map<String, Object> params)throws Exception{
		LOG.debug("HospitalAdaptor getHosMessageInfo ");
		String result ="";
		String fid = params.get("fid").toString();
		if(fid ==null || fid.equals("")){
			LOG.error("getHosMessageInfo get fid error ,fid is null or empty,fid:" + fid);
			throw new Exception("getHosMessageInfo get fid error ,fid is null or empty,fid:" + fid);
		}else{
			 adapterInterface = fidToimpClass.get(fid);
			if ((adapterInterface ==null || adapterInterface.equals("") )) {
				adapterInterface = fidToimpClass.get("default");
				LOG.warn("getHosMessageInfo get default impClass ,impClass:" + adapterInterface);
			}
			return adapterInterface.getHosMessage(params);
		}

	}

	public RetInfo trans(Map<String, Object > params){

		String fid = params.get("fid").toString();
		if(fid ==null || fid.equals("")){
			LOG.error("MqDispatcherCommon trans method get fid error ,fid is null or empty,fid:" + fid);
			retInfo.setRetCode(0);
			retInfo.setContent("MqDispatcherCommon trans method get fid error ,fid is null or empty,fid:" + fid);
		}else{
			 adapterInterface = fidToimpClass.get(fid);
			if ((adapterInterface ==null || adapterInterface.equals("") )) {
				adapterInterface = fidToimpClass.get("default");
				LOG.warn("trans get default impClass ,impClass:" + adapterInterface);
			}

            try {
                retInfo =  adapterInterface.trans(params);
            }catch (Exception e){
                LOG.error("MqDispatcherCommon trans method  call trans is error!" + e.getMessage());
                retInfo.setRetCode(0);
                retInfo.setContent("MqDispatcherCommon trans method  call trans is error!" + e.getMessage());
            }

		}
		return retInfo;
	}
}
